package com.ryan.mdp.service;

import java.util.Iterator;
import java.util.List;

import javax.jms.ConnectionFactory;
import javax.jms.Session;

import org.aopalliance.aop.Advice;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.aop.framework.ProxyFactory;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.BeanFactoryAware;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.ListableBeanFactory;
import org.springframework.jms.listener.AbstractMessageListenerContainer;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
import org.springframework.jms.listener.SimpleMessageListenerContainer;
import org.springframework.transaction.PlatformTransactionManager;

import com.ryan.mdp.advice.AdviceLoader;
import com.ryan.mdp.json.JsonMessageConverter;

public class MdpServiceFactoryBean implements FactoryBean, InitializingBean, BeanFactoryAware, BeanClassLoaderAware,
        DisposableBean {
    private static int CONCURRENCY_DEFAULT_INIT = 1;
    private static int CONCURRENCY_DEFAULT_MAX = 5;
    private final Logger logger = LoggerFactory.getLogger(getClass());

    private BeanFactory beanFactory;

    private ClassLoader classLoader;

    private Class<AbstractMessageListenerContainer> containerClass;

    private AbstractMessageListenerContainer container;

    // MDP server name
    private String server;

    private Class<?> interfaze;

    private String[] destinations;

    private boolean pubSubDomain;

    private String concurrency;
    /**
     * 消费消息时是否开启事务，true时发生异常时消息回滚，false时发生异常不回滚
     */
    private boolean transacted = false;

    private String selector;

    private int ttl;

    private Object target;

    private String targetBeanName;

    private ConnectionFactory connectionFactory;

    public Object getObject() throws Exception {
        return container;
    }

    public void afterPropertiesSet() throws Exception {
        container = containerClass.newInstance();
        container.setConnectionFactory(connectionFactory);
        container.setPubSubDomain(pubSubDomain);
        container.setDestinationName(destinations[0]);
        if (StringUtils.isNotEmpty(selector)) {
            container.setMessageSelector(selector);
        }
        MdpInvokerServiceExporter mdpExporter = new MdpInvokerServiceExporter();
        if (StringUtils.isNotEmpty(targetBeanName)) {
            target = beanFactory.getBean(targetBeanName);
        }
        target = proxyForTarget(interfaze, target);
        int[] consumers = parseConcurrency(concurrency);
        if (containerClass.isAssignableFrom(DefaultMessageListenerContainer.class)) {
            DefaultMessageListenerContainer defaultContainer = (DefaultMessageListenerContainer) container;
            defaultContainer.setConcurrentConsumers(consumers[0]);
            defaultContainer.setMaxConcurrentConsumers(consumers[1]);
            if (isTransacted()) {
                setTransactionManager(defaultContainer);
            }
        } else if (containerClass.isAssignableFrom(SimpleMessageListenerContainer.class)) {
            ((SimpleMessageListenerContainer) container).setConcurrentConsumers(consumers[0]);
        }
        mdpExporter.setService(target);
        mdpExporter.setServerName(getServer());
        mdpExporter.setServiceInterface(interfaze);
        mdpExporter.setBeanClassLoader(classLoader);
        mdpExporter.setMessageConverter(new JsonMessageConverter());
        mdpExporter.afterPropertiesSet();
        container.setMessageListener(mdpExporter);
        container.setSessionTransacted(isTransacted());
        container.setSessionAcknowledgeMode(isTransacted() ? Session.SESSION_TRANSACTED : Session.AUTO_ACKNOWLEDGE);
        container.afterPropertiesSet();
        logger.info("exporting mdp service: " + interfaze.getCanonicalName());
    }

    /**
     * 为JTA事务准备TransactionManager
     * 
     * @param defaultContainer
     */
    private void setTransactionManager(DefaultMessageListenerContainer defaultContainer) {
        ListableBeanFactory appContext = (ListableBeanFactory) beanFactory;
        String[] transManagerNames = appContext.getBeanNamesForType(PlatformTransactionManager.class);
        if (transManagerNames != null && transManagerNames.length > 0) {
            PlatformTransactionManager transManager = (PlatformTransactionManager) appContext
                    .getBean(transManagerNames[0]);
            defaultContainer.setTransactionManager(transManager);
        }
    }

    private Object proxyForTarget(Class<?> interfaze, Object target) {
        List<Advice> serviceAdvices = AdviceLoader.loadServiceAdvices();
        if (serviceAdvices.isEmpty()) {
            return target;
        }
        ProxyFactory proxyFactory = new ProxyFactory(target);
        proxyFactory.addInterface(getInterfaze());
        for (Iterator<Advice> iterator = serviceAdvices.iterator(); iterator.hasNext();) {
            Advice advice = iterator.next();
            proxyFactory.addAdvice(advice);
        }
        return proxyFactory.getProxy(classLoader);
    }

    public Class<AbstractMessageListenerContainer> getObjectType() {
        return AbstractMessageListenerContainer.class;
    }

    public boolean isSingleton() {
        return true;
    }

    public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
        this.beanFactory = beanFactory;
    }

    public void setBeanClassLoader(ClassLoader classLoader) {
        this.classLoader = classLoader;
    }

    public void destroy() throws Exception {
        if (container != null) {
            container.destroy();
        }
        AdviceLoader.unloadServiceAdvices();
    }

    public Class<AbstractMessageListenerContainer> getContainerClass() {
        return containerClass;
    }

    public void setContainerClass(Class<AbstractMessageListenerContainer> containerClass) {
        this.containerClass = containerClass;
    }

    public String getServer() {
        if (StringUtils.isEmpty(server) && target != null) {
            if (AopUtils.isAopProxy(target)) {
                server = AopUtils.getTargetClass(target).getName();
            } else {
                server = target.getClass().getName();
            }
        }
        return server;
    }

    public void setServer(String server) {
        this.server = server;
    }

    public Class<?> getInterfaze() {
        return interfaze;
    }

    public void setInterfaze(Class<?> interfaze) {
        this.interfaze = interfaze;
    }

    public String[] getDestinations() {
        return destinations;
    }

    public void setDestinations(String[] destinations) {
        this.destinations = destinations;
    }

    public boolean isPubSubDomain() {
        return pubSubDomain;
    }

    public void setPubSubDomain(boolean pubSubDomain) {
        this.pubSubDomain = pubSubDomain;
    }

    public String getConcurrency() {
        return concurrency;
    }

    public void setConcurrency(String concurrency) {
        this.concurrency = concurrency;
    }

    public int getTtl() {
        return ttl;
    }

    public void setTtl(int ttl) {
        this.ttl = ttl;
    }

    public ConnectionFactory getConnectionFactory() {
        return connectionFactory;
    }

    public void setConnectionFactory(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }

    public String getSelector() {
        return selector;
    }

    public void setSelector(String selector) {
        this.selector = selector;
    }

    public boolean isTransacted() {
        return transacted;
    }

    public void setTransacted(boolean transacted) {
        this.transacted = transacted;
    }

    public Object getTarget() {
        return target;
    }

    public void setTarget(Object target) {
        this.target = target;
    }

    public String getTargetBeanName() {
        return targetBeanName;
    }

    public void setTargetBeanName(String targetBeanName) {
        this.targetBeanName = targetBeanName;
    }

    private int[] parseConcurrency(String concurrency) {
        if (StringUtils.isEmpty(concurrency)) {
            return new int[] { CONCURRENCY_DEFAULT_INIT, CONCURRENCY_DEFAULT_MAX };
        }
        int separatorIndex = concurrency.indexOf('-');
        if (separatorIndex != -1) {
            int[] result = new int[2];
            result[0] = Integer.parseInt(concurrency.substring(0, separatorIndex));
            result[1] = Integer.parseInt(concurrency.substring(separatorIndex + 1, concurrency.length()));
            return result;
        } else {
            return new int[] { Integer.parseInt(concurrency), Integer.parseInt(concurrency) };
        }
    }
}
